package com.foreveross.crawl.workers;

import java.io.File;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Set;

import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.log4j.Logger;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.UnableToInterruptJobException;

import com.foreveross.GlobalMapOperatorBeta1;
import com.foreveross.crawl.CrawlContext;
import com.foreveross.crawl.CrawlEngineMonitor;
import com.foreveross.crawl.adapter.CrawlAdapterFactory;
import com.foreveross.crawl.adapter.IChannelAdapter;
import com.foreveross.crawl.application.impl.task.TaskOperateApplicationImpl;
import com.foreveross.crawl.common.cfg.PropertyUtils;
import com.foreveross.crawl.common.cfg.system.EnginePropertiesLoader;
import com.foreveross.crawl.common.cfg.system.SystemPropertiesLoader;
import com.foreveross.crawl.common.domain.BaseHBaseEntity;
import com.foreveross.crawl.common.domain.GrapSourceDataEntity;
import com.foreveross.crawl.common.em.TaskState;
import com.foreveross.crawl.common.exception.BusinessException;
import com.foreveross.crawl.common.exception.CustomRuntimeException;
import com.foreveross.crawl.common.exception.DefindSuccessException;
import com.foreveross.crawl.common.exception.INotRetryException;
import com.foreveross.crawl.common.exception.TaskStateException;
import com.foreveross.crawl.common.exception.self.SourceDataParseNullException;
import com.foreveross.crawl.common.exception.self.SourceWebPageNullException;
import com.foreveross.crawl.dto.TaskModelDto;
import com.foreveross.taskservice.common.TaskStatusEnum;
import com.foreveross.taskservice.common.bean.TaskModel;
import com.foreveross.taskservice.common.model.AbstractAction;
import com.foreveross.taskservice.common.model.TaskAction;
import com.foreveross.util.DateUtil;
import com.foreveross.util.NodeInfoPropertyUtil;
import com.foreveross.util.PropertesCustom;

/**
 * 默认的工作者
 * 
 * @author keman
 * 
 */
public class DefaultQueueWorkerBeta2 extends AbstractTaskWorker {
	private static Logger logger = Logger.getLogger(DefaultQueueWorkerBeta2.class);

	public DefaultQueueWorkerBeta2() {
		super();
	}
	public void execute(JobExecutionContext context)throws JobExecutionException {
		TaskModel task = null;
		long channelId = 0;
		long startTime = System.currentTimeMillis();
		String taskId = "";
		try {
			// 1、 获取队列信息
			task = (TaskModel) context.getJobDetail().getJobDataMap().get(CrawlContext.CRAWL_TASK);
			taskId = task.getId();
			channelId= context.getJobDetail().getJobDataMap().getLong(CrawlContext.CRAWL_CHANNEL_ID);

			IChannelAdapter adapter = null;
			try {
 				adapter = CrawlAdapterFactory.buildAdaptor(task);
				this.executeTask(adapter, task, channelId);
				// 5、成功记录
				this.success(task, startTime, "任务执行成功!");
			} catch (Exception e) {
				this.dealException(channelId, task, startTime, e);
			} finally {
				if(adapter != null){//2014-07-28 释放HttpClient连接
					adapter.destory();
					adapter = null;
				}
				task.setEndTime(new Date());
				//TODO 先清空轨迹,MQ及任务服务那边撑不住
				task.getActions().clear();
				//TODO 暂时将该动作设置为TASK_VALIDATE
				this.logAction(task, TaskAction.TASK_VALIDATE, "执行任务的节点SN:" + NodeInfoPropertyUtil.getValueByProperty(NodeInfoPropertyUtil.NODE_SN), false);
				iCrawlNodeBaseTaskApplication.roolbackTaskQueueRecordEntity(task);
			}
		} catch (Exception e1) {
			try {
				this.dealException(channelId, task, startTime, e1);
			} catch (Exception e) {
				logger.error(e);
			}
			CrawlEngineMonitor.getInstance().allFailTaskCount();
		} finally {
			CrawlEngineMonitor.getInstance().allTaskCount();
			task = null;
		}
	}

	private void success(TaskModel task, long startTime,
			String remark) throws BusinessException {
		task.setStatus(TaskStatusEnum.TASK_SUCCESS.getStatus());
		task.setRemarks(remark);
		CrawlEngineMonitor.getInstance().allDoneTaskCount();
		
//		this.updateTaskState(task.getChannel().getId(), task.getId(), TaskState.FINISHED.getType());
		
		dealTaskModel(task, startTime, TaskState.FINISHED.getType());
		this.recordTracks(task.getChannel().getId(), task.getId(), remark);
		
		logger.info(remark);
	}
	
	private void success(int errorCode, TaskModel task, long startTime,
			String remark) throws BusinessException {
		this.success(task, startTime, remark);
		task.setStatus(errorCode);
	}

	/**
	 * 循环执行抓取
	 * <p>
	 * 这里进行了一次业务上的验证，如果适配器里验证不通过，则循环指定次数抓取，若有抓取“正确”的结果则停止循环,解析该结果
	 * 
	 * @param task
	 * @param adapter
	 * @param record
	 * @throws Exception
	 */
	protected void executeTask(IChannelAdapter adapter,
			TaskModel task, long channelId)
			throws Exception {
		// 3、抓取数据
		Object fetchObject = this.fetch(adapter, task, channelId);
		// 4、解析数据
		this.parseData(adapter, task, fetchObject);
	}

	/**
	 * 解析数据
	 * 
	 * @param adapter
	 * @param record
	 * @param obj
	 * @throws Exception
	 */
	protected void parseData(IChannelAdapter adapter,
			TaskModel record, Object obj) throws Exception {
		List<BaseHBaseEntity> list = null;
		List<Object> results = null;
		long stime = System.currentTimeMillis();
		try {// 调用适配器的解析方法（以后优化为另一个解析模块）
			logAction(record, TaskAction.TASK_PARASE, " 解析数据", true);
			results = adapter.paraseToVo(obj);
			list = this.convertObject(results, record);
			record.setParseDurationTime(System.currentTimeMillis() - stime);

			if (list == null || list.isEmpty()) {
				throw new SourceDataParseNullException("将源网页数据解析出的模型数据为空!");
			}
			record.setCrawlCount(list.size());
			stime = System.currentTimeMillis();
			logger.info("准备存储解析数据");
			//是否入库
			if(PropertyUtils.getBooleanProperty(PropertesCustom.SYSTEM_DB_ISSTORAGE, SystemPropertiesLoader.FILE_NAME)){
				this.recordTracks(record.getChannel().getId(), record.getId(), "数据解析，准备存储入库");
				planeInfoService.creates(list);
			} else {
				this.recordTracks(record.getChannel().getId(), record.getId(), "数据解析，准备将解析结果存储在本地文件");
				this.iTaskOperateApplication.dealParseJson(list, record);
			}
			this.recordTracks(record.getChannel().getId(), record.getId(), "存储数据花费时间  "+(System.currentTimeMillis() - stime)+ "毫秒");
			logger.info("存储解析数据花费时间  "+(System.currentTimeMillis() - stime)+ "毫秒");
			// 记录存储时间
			record.setStoreDurationTime(System.currentTimeMillis() - stime);
		} finally {
			list = null;
			results = null;
		}
	}

	/**
	 * 控制扩展，一个任务抓取失败后的重复时的间隔时间，采用不同渠道不同配置
	 * @return
	 */
	protected long getEngineFetchInterval(TaskModel task){
		long fetchInterval=PropertyUtils.getIntProperty(
				EnginePropertiesLoader.ENGINE_TASK_CRAWL_FETCHINTERVAL+".channel"+task.getGrabChannelId(), EnginePropertiesLoader.FILE_NAME, -1);
		return fetchInterval > 0 ? fetchInterval : PropertyUtils.getIntProperty(
				EnginePropertiesLoader.ENGINE_TASK_CRAWL_FETCHINTERVAL, EnginePropertiesLoader.FILE_NAME,10000);
	}
	/**
	 * 循环抓取数据
	 * <p>
	 * 当抓取不成功时，进行多次抓取。
	 * <p>
	 * 在抓取前，先取得代理IP，每一次抓取完成后，要回滚通知代理ip服务
	 * <p>
	 * 用一个内部类包装抓取的结果，此方法全程不会向外抛出异常
	 * @param task
	 * @param adapter
	 * @param record
	 * @return
	 * @throws Exception
	 */
	protected FetchResult fetchData(IChannelAdapter adapter,TaskModel task, long channelId){
		FetchResult result=new FetchResult();
		int count = PropertyUtils.getIntProperty(EnginePropertiesLoader.ENGINE_TASK_CRAWL_MAXREPEATCOUNT, EnginePropertiesLoader.FILE_NAME,10);
		Long stime = System.currentTimeMillis();
		String taskId = task.getId();
		long fetchInterval = getEngineFetchInterval(task);
		int times = 1;// 执行次数
		while (count-- > 0) {
			try {
				logger.info(String.format("渠道[%s] 倒数第[%s]执行任务",
						task.getGrabChannel(), count));
				this.recordTracks(channelId, taskId, String.format(
						"%s第%s次执行任务", task.getGrabChannel(), times));
				times++;

				result.clear();
				// 适配器抓取数据
				logAction(task, TaskAction.TASK_FETCH, " 进入网站", true);
				Object obj = adapter.fetch(null);
				result.setFetchResult(obj);
				logAction(task, TaskAction.TASK_VALIDATE, " 执行验证", true);
				if (adapter.validateFetch(obj)) {// 验证通过了才解析
					break;
				}
			} catch (Exception e) {
				result.setException(e);// 将异常记录
				logAction(task, TaskAction.ERROR, "异常类型: " + e.getClass(), e,
						true);
				if (count <= 0) {// 次数用光
					break;
				}
				if (this.isLegalException(e)) {
					break;
				}
				try {
					Thread.sleep(fetchInterval);// 休息一下下
					adapter.switchProxyip();// 异常时，重复下一次前切换好IP
				} catch (InterruptedException e1) {
					this.logAction(task, TaskAction.ERROR, "", e1, true);
				}
			} finally {
				task.setGrabTimes(task.getGrabTimes() + 1);// 记录抓取次数
			}
		}
		task.setCrawlDurationTime(System.currentTimeMillis() - stime);
		return result;
	}
	/**
	 * 判断是不是属于一个合法的异常，如 没票 没航班 卖光等等异常
	 * @param e
	 * @return
	 */
	private boolean isLegalException(Exception e){
		if (e instanceof DefindSuccessException || e instanceof INotRetryException) {
			return true;
		}
		return false;
	}
	
	
	/**
	 * 抓取数据
	 * <p>
	 * 1、调用抓取方法循环抓取<br/>
	 * 2、将抓取的源网页数据存入hbase<br/>
	 * 3、记录hbase的rowkey，源数据大小到日志记录<br/>
	 * 
	 * @param task
	 * @param adapter
	 * @param record
	 * @param channel
	 * @return
	 * @throws Exception
	 */
	protected Object fetch(IChannelAdapter adapter, TaskModel task,
			long channelId) throws Exception {
		FetchResult f = null;
		try {
			f = this.fetchData(adapter, task, channelId);

			// 将一切记录下来之后，如果有异常，要抛出
			if (f.getException() != null) {
				throw f.getException();
			}
			if (f.getFetchResult() == null && f.getException() == null) {
				throw new SourceWebPageNullException("无法捕获到镜像文件和任何异常");
			}
		} finally {
			saveSourceFile(adapter, task);
		}
		return f.getFetchResult();
	}
	
	private void saveSourceFile(IChannelAdapter adapter, TaskModel task) {
		String temp;
		GrapSourceDataEntity sourceDataEntity;
		StringBuffer condition;
		// 是否入库
		if (PropertyUtils.getBooleanProperty(PropertesCustom.SYSTEM_DB_ISSTORAGE,SystemPropertiesLoader.FILE_NAME) ) {
			try {
				temp = adapter.getSourceWebPageContents();
				if(StringUtils.isNotEmpty(temp)){
					//源网页数据表的rowKey
					//渠道(往|返)航线航班时间(yyyyMMdd)抓取时间(yyyyMMddHHmmss)五位随机数
					//作为参数，我们仅仅提供前面的，后面的五位随机数就不需要提供了
					
					condition=new StringBuffer()
						.append(StringUtils.leftPad(String.valueOf(task.getAttachHbaseKey()), 5, '0'))
						.append("_")
						.append(task.getFromCity()+task.getToCity())
						.append("_")
						.append(task.getFlightDate().replaceAll("\\s", ""))
						.append("_")
						.append(DateFormatUtils.format(task.getStartTime(), "yyyyMMddHHmmss"));
					
					sourceDataEntity=new GrapSourceDataEntity(condition.toString(), temp, null);
					this.recordTracks(task.getChannel().getId(), task.getId(), String.format("准备将%s网页文件存入hbase", task.getGrabChannel()));
					grapSourceDataService.save(sourceDataEntity);
					this.recordTracks(task.getChannel().getId(), task.getId(), String.format("%s网页文件存入hbase,数据长度:%s", task.getGrabChannel(), temp.length()));
					logger.info(String.format("渠道[%s]网页文件存入hbase,数据长度[%s]",
							task.getGrabChannel(), temp.length()));
					
					// 保存源始的rowkey到记录表，以用于追逆
					task.setSourceRowkey((String) sourceDataEntity.getId());
					task.setSourceDataSize(temp.length());
					task.setByteLength(temp.getBytes().length);
				}
			} catch (Exception e1) {
				this.recordTracks(task.getChannel().getId(), task.getId(),
						String.format("%s记录抓取源数据出错", task.getGrabChannel()));
				logger.error("渠道[" + task.getGrabChannel() + "]记录抓取源数据出错"
						+ ExceptionUtils.getFullStackTrace(e1));
			} finally {
				temp = null;
				sourceDataEntity = null;
			}
		}
	}	
	
	/**
	 * 将Object转换为我们需要的业务对象PlaneInfoEntity对象
	 * 
	 * @param results
	 * @param record
	 * @return
	 */
	private List<BaseHBaseEntity> convertObject(List<Object> results,
			TaskModel record) {
		List<BaseHBaseEntity> list = new ArrayList<BaseHBaseEntity>();
		BaseHBaseEntity entity = null;
		if (results != null) {
			for (Object o : results) {
				if (o instanceof BaseHBaseEntity) {
					entity = (BaseHBaseEntity) o;
					entity.setSourceRowkey(record.getSourceRowkey()); // 记录下源网页存储表的ID，建立了主外键关联关系
						list.add(entity);
					}
				}
			}
		return list;
	}

	@Override
	public void interrupt() throws UnableToInterruptJobException {}

	private void logAction(TaskModel model,int type,String remark, boolean recordToLocal){
		if(PropertyUtils.getBooleanProperty(EnginePropertiesLoader.ENGINE_TASK_CRAWL_ACTION_LOG, EnginePropertiesLoader.FILE_NAME)){
			TaskAction action=new TaskAction(type, new Date().getTime(),remark);
			model.addAction(action);
			Set<AbstractAction> actionSet = model.getActions();
			if (recordToLocal) {
				// 2014-07-15 将任务轨迹缓存到本地
				this.recordTracks(model.getChannel().getId(), model.getId(), remark);
			}
		}
	}
	
	private void logAction(TaskModel model,int type,String remark, Exception e){
		if(PropertyUtils.getBooleanProperty(EnginePropertiesLoader.ENGINE_TASK_CRAWL_ACTION_LOG, EnginePropertiesLoader.FILE_NAME) == true){
			TaskAction action=new TaskAction(type, new Date().getTime(),remark+System.getProperty("line.separator")+exceptionToList(e));
			model.addAction(action);
		}
		logger.error(String.format("异常任务信息: 渠道名称%s,起飞城市%s,到达城市%s,去程时间%s,返程时间%s", model.getChannel().getName(), model.getFromCity(), model.getToCity(), model.getFlightDate(), model.getReturnGrabDate()));
	}
	
	private void logAction(TaskModel model,int type,String remark, Exception e, boolean isRecord){
		this.logAction(model, type, remark, e);
		if(isRecord){
			this.recordTracks(model.getChannel().getId(), model.getId(), remark+System.getProperty("line.separator")+exceptionToList(e));
		}
	}
	
	/**
	 * 内部类，fetch结果集
	 * @author Administrator
	 *
	 */
	class FetchResult{
		private Object fetchResult;//抓取的结果（网页镜像文件）
		private Exception exception;//抓取中途抛出的异常
		public Object getFetchResult() {
			return fetchResult;
		}
		public void setFetchResult(Object fetchResult) {
			this.fetchResult = fetchResult;
		}
		public Exception getException() {
			return exception;
		}
		public void setException(Exception exception) {
			this.exception = exception;
		}
		
		public void clear(){
			this.fetchResult=null;
			this.exception=null;
		}
	}
	
	/**
	 * 更新Map缓存中的任务状态
	 * @param channel
	 * @param taskId
	 * @param state
	 */
	private void updateTaskState(Long channel, String taskId, String state){
		try {
			this.iTaskModelEntityApplication.updateTaskState(taskId, state);
		} catch (Exception e) {
			logger.error(e.toString());
		}
	}
	
	private void dealTaskModel(TaskModel model, long startTime, String state){
		Long channelId = model.getChannel().getId();
		String taskId = model.getId();
		try {
			GlobalMapOperatorBeta1.updateTaskEndDateAndExcuteTime(channelId, taskId, System.currentTimeMillis() - startTime, state);
		} catch (Exception e) {
			logger.error(e);
		}
		TaskModelDto dto = GlobalMapOperatorBeta1.getTaskByChannelAndTask(channelId, taskId);
		if(dto != null){
//			dto.setTaskState(state);
			//将任务信息存入H2
			this.iTaskModelEntityApplication.addTaskModel(dto);
		}
		try {
			//减少一个正在做的任务
			GlobalMapOperatorBeta1.decreaseTaskDoingNumber(channelId);
		} catch (BusinessException e) {
			logger.error(e);
		}
		//清除缓存中的任务信息
		GlobalMapOperatorBeta1.removeTaskByChannelAndTask(channelId, taskId);
	}
	
	/**  
	 * @Description: 记录任务动作轨迹
	 * @param channel
	 * @param taskId
	 * @param tracks
	 * @author luofangyi
	 * @date 2014-7-15 上午10:50:49 
	 */ 
	private void recordTracks(Long channel, String taskId, String tracks){
		try {
//			this.iTaskModelEntityApplication.recordTracksOfTask(taskId, tracks);
			GlobalMapOperatorBeta1.recordTracksOfTask(channel, taskId, DateUtil.date2String(new Date(), "HH:mm:ss") + " " + tracks);
		} catch (TaskStateException e) {
			logger.error(e.toString());
		}
	}
	
	/**  
	 * @Description: 异常信息转字符串
	 * @param e
	 * @return
	 * @author luofangyi
	 * @date 2014-7-17 下午2:32:03 
	 */ 
	private String exceptionToList(Exception e){
		StringBuffer sbuilder = new StringBuffer("");
		try {
			sbuilder.append(e.getMessage());
			sbuilder.append(System.getProperty("line.separator"));
			boolean isShow = PropertyUtils.getBooleanProperty(PropertesCustom.SYSTEM_ISSHOW_FOREVEROSS_EXCEPTION, SystemPropertiesLoader.FILE_NAME);
			for(StackTraceElement stack : e.getStackTrace()){
				if(stack.getClassName().contains("com.foreveross") && !isShow){
					sbuilder.append(String.format("%s.%s(%s)",  stack.getClassName(),stack.getMethodName(),stack.getLineNumber())).append(System.getProperty("line.separator"));
				} else if(isShow) {
					sbuilder.append(String.format("%s.%s(%s)",  stack.getClassName(),stack.getMethodName(),stack.getLineNumber())).append(System.getProperty("line.separator"));
				}
			}
		} catch (Exception e1) {
			logger.error(e1);
		}
		return sbuilder.toString();
	}
	
	private void dealException(Long channelId, TaskModel task, long startTime, Exception e) throws Exception{
		if(e instanceof DefindSuccessException){
			DefindSuccessException successException = (DefindSuccessException)e;
			this.success(successException.getErrorCode(), task, startTime, this.exceptionToList(successException));
		} else {
			if (e instanceof BusinessException) {
				task.setStatus(((BusinessException)e).getErrorCode());
			} else if (e instanceof CustomRuntimeException) {
				task.setStatus(((CustomRuntimeException)e).getErrorCode());
			} else {
				task.setStatus(TaskStatusEnum.TASK_OTHER_FAIL.getStatus());
			}
			task.setRemarks(e.getMessage());
			logAction(task, TaskAction.ERROR, "异常类型: " + e.getClass(), e, true);
//			this.updateTaskState(channelId, task.getId(),
//					TaskState.EXCEPTION.getType());
			dealTaskModel(task, startTime, TaskState.EXCEPTION.getType());
			CrawlEngineMonitor.getInstance().allFailTaskCount();
		}
		logger.error(String.format("渠道[%s]异常:%s", task.getGrabChannel(),
						e.getMessage()), e);
		if (logger.isInfoEnabled()) {
			e.printStackTrace();
		}
		
	}
	
	public static void main(String[] args) throws Exception{
		File file = new File("D:/webcrawl");
		file.list();
		while(true){
		for(final String name : file.list()){
//			System.out.println(name);
			
			new Thread(new Runnable() {
				
				@Override
				public void run() {
					// TODO Auto-generated method stub
					
					try {
						new TaskOperateApplicationImpl().dealZip("D:/webcrawl/" + name, name, 23l);
					} catch (Exception e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
			}).start();
//			Thread.sleep(2000);
		}
		Thread.sleep(3000);
		}
	}
}
